package com.zhiqian.community.event;

import com.alibaba.fastjson.JSONObject;
import com.aliyun.oss.ClientException;
import com.aliyun.oss.OSS;
import com.aliyun.oss.OSSException;
import com.aliyun.oss.model.PutObjectResult;
import com.zhiqian.community.entity.Event;
import com.zhiqian.community.service.ElasticsearchService;
import com.zhiqian.community.service.MessageService;
import com.zhiqian.community.util.CommunityConstant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.Future;

/**
 * 事件消费者
 */
@Component
public class EventConsumer implements CommunityConstant {

    private static final Logger logger = LoggerFactory.getLogger(EventConsumer.class);
    @Autowired
    private MessageService messageService;
    @Autowired
    private ElasticsearchService elasticsearchService;
    @Autowired
    private ThreadPoolTaskScheduler taskScheduler;
    @Autowired
    private OSS ossClient;
    @Value("${wk.image.storage}")
    private String imageStorage;
    @Value("${wk.image.command}")
    private String command;
    @Value("${share-bucket}")
    private String shareBucket;

    @KafkaListener(topics = {TOPIC_LIKE})
    public void likeHandler(ConsumerRecord record) {
        Event event = getEvent(record);
        if (event != null) {
            messageService.sendNotification(event);
        }
    }

    @KafkaListener(topics = {TOPIC_COMMENT})
    public void commentHandler(ConsumerRecord record) {
        Event event = getEvent(record);
        if (event != null) {
            messageService.sendNotification(event);
        }
    }

    @KafkaListener(topics = {TOPIC_FOLLOW})
    public void followHandler(ConsumerRecord record) {
        Event event = getEvent(record);
        if (event != null) {
            messageService.sendNotification(event);
        }
    }

    @KafkaListener(topics = {TOPIC_PUBLISH})
    public void publishHandler(ConsumerRecord record) {
        Event event = getEvent(record);
        if (event != null) {
            elasticsearchService.insertDiscussPost(event.getEntityId());
        }
    }

    @KafkaListener(topics = {TOPIC_REMOVED})
    public void removedHandler(ConsumerRecord record) {
        Event event = getEvent(record);
        if (event != null) {
            elasticsearchService.deleteDiscussPostById(event.getEntityId());
        }
    }

    @KafkaListener(topics = {TOPIC_SHARE})
    public void shareHandler(ConsumerRecord record) {
        Event event = getEvent(record);
        if (event != null) {
            Map<String, Object> data = event.getData();
            String htmlUrl = (String) data.get("htmlUrl");
            String fileName = (String) data.get("fileName");
            String suffix = (String) data.get("suffix");
            String cmd = command + htmlUrl + " " + imageStorage + "/" + fileName + suffix;
            try {
                Runtime.getRuntime().exec(cmd);
                logger.info("生成长图成功：" + cmd);
//                UploadToOSS task = new UploadToOSS(fileName, suffix);
                //生成定时任务：上传长图
//                ScheduledFuture<?> future = taskScheduler.scheduleAtFixedRate(task, 500);
//                task.setFuture(future);
            } catch (IOException e) {
                logger.error("生成长图失败：" + cmd, e);
            }
        }
    }

    @KafkaListener(topics = {TOPIC_INFORM})
    public void informHandler(ConsumerRecord record){
        Event event = getEvent(record);
        if (event!=null){
            messageService.sendInform(event);
        }
    }

    private Event getEvent(ConsumerRecord record) {
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空");
            return null;
        }
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {
            logger.error("消息格式错误");
        }
        return event;
    }

    class UploadToOSS implements Runnable {
        private String fileName;
        private String suffix;
        private Future future;
        //任务开始时间
        private long startTime;
        //上传的次数
        private int uploadTimes;

        public UploadToOSS(String fileName, String suffix) {
            this.fileName = fileName;
            this.suffix = suffix;
            this.startTime = System.currentTimeMillis();
        }

        public void setFuture(Future future) {
            this.future = future;
        }

        @Override
        public void run() {
            //生成长图失败
            if (System.currentTimeMillis() - startTime > 30000) {
                logger.error("执行时间过长，终止上传任务：" + fileName);
                future.cancel(true);
                return;
            }
            if (uploadTimes > 3) {
                logger.error("图片上传重试次数过多，终止上传任务，请检查网络或OSS服务：" + fileName);
                future.cancel(true);
                return;
            }
            File file = new File(imageStorage + "/" + fileName + suffix);
            if (file.exists()) {
                try {
                    //此处等一段时间，文件存在但是可能还没有生成完毕并不完整，如果直接上传会出现空文件
                    Thread.sleep(500);
                    logger.info(String.format("开始第%d次上传 [%s]", ++uploadTimes, fileName));
                    PutObjectResult result = ossClient.putObject(shareBucket, fileName + suffix, file);
                    String eTag = result.getETag();
                    logger.info("长图上传OSS成功：" + eTag);
                    future.cancel(true);
                }catch (OSSException oe) {
                    logger.error(String.format("第%d次上传失败 [%s]",uploadTimes,fileName));
                    logger.error("上传OSS请求被拒绝");
                    logger.error("错误信息:" + oe.getErrorMessage());
                    logger.error("错误码:" + oe.getErrorCode());
                    logger.error("请求 ID:" + oe.getRequestId());
                    logger.error("主机 ID:" + oe.getHostId());
                } catch (ClientException ce) {
                    logger.error(String.format("第%d次上传失败 [%s]",uploadTimes,fileName));
                    logger.error("客户端异常：请检查网络",ce.getErrorMessage());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }else {
                logger.info("正在等待图片生成："+fileName);
            }
        }
    }
}
